/*-------------------------------------------------------------------------
 * relation.c
 *	   PostgreSQL 逻辑复制关系映射缓存
 *
 * Copyright (c) 2016-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/logical/relation.c
 *
 * NOTES
 *	  此文件中的例程主要涉及将本地复制目标关系的属性映射到
 *	  它们的远程对应物的属性上。
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/table.h"
#include "catalog/namespace.h"
#include "catalog/pg_subscription_rel.h"
#include "executor/executor.h"
#include "nodes/makefuncs.h"
#include "replication/logicalrelation.h"
#include "replication/worker_internal.h"
#include "utils/inval.h"


static MemoryContext LogicalRepRelMapContext = NULL;

static HTAB *LogicalRepRelMap = NULL;

/*
 * 分区映射 (LogicalRepPartMap)
 *
 * 当分区表作为复制目标时，复制操作实际上是在它的叶子分区上执行的，这要求这些分区也必须映射到远程关系。父项的条目
 * (LogicalRepRelMapEntry) 不能直接用于所有分区，因为单个分区可能具有不同的属性数量，这意味着
 * 必须单独维护每个分区到远程关系属性的属性映射。
 */
static MemoryContext LogicalRepPartMapContext = NULL;
static HTAB *LogicalRepPartMap = NULL;
typedef struct LogicalRepPartMapEntry
{
	Oid			partoid;		/* LogicalRepPartMap的键 */
	LogicalRepRelMapEntry relmapentry;
} LogicalRepPartMapEntry;

/*
 * Relcache失效回调用于我们的关系映射缓存。
 */
static void fc_logicalrep_relmap_invalidate_cb(Datum fc_arg, Oid fc_reloid)
{
	LogicalRepRelMapEntry *fc_entry;

	/* 只是为了确保。 */
	if (LogicalRepRelMap == NULL)
		return;

	if (fc_reloid != InvalidOid)
	{
		HASH_SEQ_STATUS fc_status;

		hash_seq_init(&fc_status, LogicalRepRelMap);

		/* TODO，使用逆查找哈希表？ */
		while ((fc_entry = (LogicalRepRelMapEntry *) hash_seq_search(&fc_status)) != NULL)
		{
			if (fc_entry->localreloid == fc_reloid)
			{
				fc_entry->localrelvalid = false;
				hash_seq_term(&fc_status);
				break;
			}
		}
	}
	else
	{
		/* 使所有缓存条目失效 */
		HASH_SEQ_STATUS fc_status;

		hash_seq_init(&fc_status, LogicalRepRelMap);

		while ((fc_entry = (LogicalRepRelMapEntry *) hash_seq_search(&fc_status)) != NULL)
			fc_entry->localrelvalid = false;
	}
}

/*
 * 初始化关系映射缓存。
 */
static void fc_logicalrep_relmap_init(void)
{
	HASHCTL		fc_ctl;

	if (!LogicalRepRelMapContext)
		LogicalRepRelMapContext =
			AllocSetContextCreate(CacheMemoryContext,
								  "LogicalRepRelMapContext",
								  ALLOCSET_DEFAULT_SIZES);

	/* 初始化关系哈希表。 */
	fc_ctl.keysize = sizeof(LogicalRepRelId);
	fc_ctl.entrysize = sizeof(LogicalRepRelMapEntry);
	fc_ctl.hcxt = LogicalRepRelMapContext;

	LogicalRepRelMap = hash_create("logicalrep relation map cache", 128, &fc_ctl,
								   HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);

	/* 监视失效事件。 */
	CacheRegisterRelcacheCallback(fc_logicalrep_relmap_invalidate_cb,
								  (Datum) 0);
}

/*
 * 释放关系映射缓存的条目。
 */
static void fc_logicalrep_relmap_free_entry(LogicalRepRelMapEntry *fc_entry)
{
	LogicalRepRelation *fc_remoterel;

	fc_remoterel = &fc_entry->remoterel;

	pfree(fc_remoterel->nspname);
	pfree(fc_remoterel->relname);

	if (fc_remoterel->natts > 0)
	{
		int			fc_i;

		for (fc_i = 0; fc_i < fc_remoterel->natts; fc_i++)
			pfree(fc_remoterel->attnames[fc_i]);

		pfree(fc_remoterel->attnames);
		pfree(fc_remoterel->atttyps);
	}
	bms_free(fc_remoterel->attkeys);

	if (fc_entry->attrmap)
		free_attrmap(fc_entry->attrmap);
}

/*
 * 在关系映射缓存中添加新条目或更新现有条目。
 *
 * 当发布者发送新的关系映射以更新我们对来自该发布者的
 * 数据的预期视图时调用。
 */
void logicalrep_relmap_update(LogicalRepRelation *fc_remoterel)
{
	MemoryContext fc_oldctx;
	LogicalRepRelMapEntry *fc_entry;
	bool		fc_found;
	int			fc_i;

	if (LogicalRepRelMap == NULL)
		fc_logicalrep_relmap_init();

	/*
	 * HASH_ENTER 返回现有条目（如果存在）或创建一个新条目。
	 */
	fc_entry = hash_search(LogicalRepRelMap, (void *) &fc_remoterel->remoteid,
						HASH_ENTER, &fc_found);

	if (fc_found)
		fc_logicalrep_relmap_free_entry(fc_entry);

	memset(fc_entry, 0, sizeof(LogicalRepRelMapEntry));

	/* 制作数据的缓存副本 */
	fc_oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
	fc_entry->remoterel.remoteid = fc_remoterel->remoteid;
	fc_entry->remoterel.nspname = pstrdup(fc_remoterel->nspname);
	fc_entry->remoterel.relname = pstrdup(fc_remoterel->relname);
	fc_entry->remoterel.natts = fc_remoterel->natts;
	fc_entry->remoterel.attnames = palloc(fc_remoterel->natts * sizeof(char *));
	fc_entry->remoterel.atttyps = palloc(fc_remoterel->natts * sizeof(Oid));
	for (fc_i = 0; fc_i < fc_remoterel->natts; fc_i++)
	{
		fc_entry->remoterel.attnames[fc_i] = pstrdup(fc_remoterel->attnames[fc_i]);
		fc_entry->remoterel.atttyps[fc_i] = fc_remoterel->atttyps[fc_i];
	}
	fc_entry->remoterel.replident = fc_remoterel->replident;
	fc_entry->remoterel.attkeys = bms_copy(fc_remoterel->attkeys);
	MemoryContextSwitchTo(fc_oldctx);
}

/*
 * 通过属性名称在TupleDesc结构中查找属性索引。
 *
 * 如果未找到，则返回-1。
 */
static int fc_logicalrep_rel_att_by_name(LogicalRepRelation *fc_remoterel, const char *fc_attname)
{
	int			fc_i;

	for (fc_i = 0; fc_i < fc_remoterel->natts; fc_i++)
	{
		if (strcmp(fc_remoterel->attnames[fc_i], fc_attname) == 0)
			return fc_i;
	}

	return -1;
}

/*
 * 报告缺失的本地关系列的名称（如果有）的错误。
 */
static void fc_logicalrep_report_missing_attrs(LogicalRepRelation *fc_remoterel,
								Bitmapset *fc_missingatts)
{
	if (!bms_is_empty(fc_missingatts))
	{
		StringInfoData fc_missingattsbuf;
		int			fc_missingattcnt = 0;
		int			fc_i;

		initStringInfo(&fc_missingattsbuf);

		while ((fc_i = bms_first_member(fc_missingatts)) >= 0)
		{
			fc_missingattcnt++;
			if (fc_missingattcnt == 1)
				appendStringInfo(&fc_missingattsbuf, _("\"%s\""),
								 fc_remoterel->attnames[fc_i]);
			else
				appendStringInfo(&fc_missingattsbuf, _(", \"%s\""),
								 fc_remoterel->attnames[fc_i]);
		}

		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg_plural("logical replication target relation \"%s.%s\" is missing replicated column: %s",
							   "logical replication target relation \"%s.%s\" is missing replicated columns: %s",
							   fc_missingattcnt,
							   fc_remoterel->nspname,
							   fc_remoterel->relname,
							   fc_missingattsbuf.data)));
	}
}

/*
 * 检查副本身份是否匹配并标记可更新标志。
 *
 * 我们允许订阅者在副本身份（更少的列）上有更严格的要求，因为这不会阻止我们找到唯一的元组。即，如果发布者具有
 * 身份 (id,timestamp) 而订阅者仅 (id)，这将不会是一个问题，但在相反的情况下就会。
 *
 * 如果发现本地副本身份不足以应用更新/删除（插入不在乎！），
 * 我们就在这里将关系条目标记为不可更新，并将实际错误留给
 * check_relation_updatable() 抛出（如果需要）。
 */
static void fc_logicalrep_rel_mark_updatable(LogicalRepRelMapEntry *fc_entry)
{
	Bitmapset  *fc_idkey;
	LogicalRepRelation *fc_remoterel = &fc_entry->remoterel;
	int			fc_i;

	fc_entry->updatable = true;

	fc_idkey = RelationGetIndexAttrBitmap(fc_entry->localrel,
									   INDEX_ATTR_BITMAP_IDENTITY_KEY);
	/* 如果没有副本身份，则回退到PK */
	if (fc_idkey == NULL)
	{
		fc_idkey = RelationGetIndexAttrBitmap(fc_entry->localrel,
										   INDEX_ATTR_BITMAP_PRIMARY_KEY);

		/*
		 * 如果没有副本身份索引且没有PK，则发布的表必须
		 * 具有副本身份FULL。
		 */
		if (fc_idkey == NULL && fc_remoterel->replident != REPLICA_IDENTITY_FULL)
			fc_entry->updatable = false;
	}

	fc_i = -1;
	while ((fc_i = bms_next_member(fc_idkey, fc_i)) >= 0)
	{
		int			fc_attnum = fc_i + FirstLowInvalidHeapAttributeNumber;

		if (!AttrNumberIsForUserDefinedAttr(fc_attnum))
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("logical replication target relation \"%s.%s\" uses "
							"system columns in REPLICA IDENTITY index",
							fc_remoterel->nspname, fc_remoterel->relname)));

		fc_attnum = AttrNumberGetAttrOffset(fc_attnum);

		if (fc_entry->attrmap->attnums[fc_attnum] < 0 ||
			!bms_is_member(fc_entry->attrmap->attnums[fc_attnum], fc_remoterel->attkeys))
		{
			fc_entry->updatable = false;
			break;
		}
	}
}

/*
 * 打开与远程关系关联的本地关系。
 *
 * 如果由于本地DDL而使Relcache映射失效，则重建该映射。
 */
LogicalRepRelMapEntry * logicalrep_rel_open(LogicalRepRelId fc_remoteid, LOCKMODE fc_lockmode)
{
	LogicalRepRelMapEntry *fc_entry;
	bool		fc_found;
	LogicalRepRelation *fc_remoterel;

	if (LogicalRepRelMap == NULL)
		fc_logicalrep_relmap_init();

	/* 搜索现有条目。 */
	fc_entry = hash_search(LogicalRepRelMap, (void *) &fc_remoteid,
						HASH_FIND, &fc_found);

	if (!fc_found)
		elog(ERROR, "no relation map entry for remote relation ID %u",
			 fc_remoteid);

	fc_remoterel = &fc_entry->remoterel;

	/* 确保我们不会泄漏relcache引用计数。 */
	if (fc_entry->localrel)
		elog(ERROR, "remote relation ID %u is already open", fc_remoteid);

	/*
	 * 在打开和锁定关系时，处理待处理的失效消息，这可能会使关系失效。
	 * 因此，如果条目当前被认为是有效的，尝试通过OID打开本地关系并
	 * 查看是否发生失效。
	 */
	if (fc_entry->localrelvalid)
	{
		fc_entry->localrel = try_table_open(fc_entry->localreloid, fc_lockmode);
		if (!fc_entry->localrel)
		{
			/* 表被重命名或删除。 */
			fc_entry->localrelvalid = false;
		}
		else if (!fc_entry->localrelvalid)
		{
			/* 请注意，我们在这里释放不再有用的锁。 */
			table_close(fc_entry->localrel, fc_lockmode);
			fc_entry->localrel = NULL;
		}
	}

	/*
	 * 如果自上次对条目加锁以来条目已被标记为无效，
	 * 按名称重新打开本地关系并重建所有派生数据。
	 */
	if (!fc_entry->localrelvalid)
	{
		Oid			fc_relid;
		TupleDesc	fc_desc;
		MemoryContext fc_oldctx;
		int			fc_i;
		Bitmapset  *fc_missingatts;

		/* 释放不再有用的 attrmap（属性映射），如果有的话。 */
		if (fc_entry->attrmap)
		{
			free_attrmap(fc_entry->attrmap);
			fc_entry->attrmap = NULL;
		}

		/* 尝试通过名称查找并锁定关系。 */
		fc_relid = RangeVarGetRelid(makeRangeVar(fc_remoterel->nspname,
											  fc_remoterel->relname, -1),
								 fc_lockmode, true);
		if (!OidIsValid(fc_relid))
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("logical replication target relation \"%s.%s\" does not exist",
							fc_remoterel->nspname, fc_remoterel->relname)));
		fc_entry->localrel = table_open(fc_relid, NoLock);
		fc_entry->localreloid = fc_relid;

		/* 检查支持的 relkind。 */
		CheckSubscriptionRelkind(fc_entry->localrel->rd_rel->relkind,
								 fc_remoterel->nspname, fc_remoterel->relname);

		/*
		 * 构建本地属性编号到远程属性编号的映射
		 * 并验证我们没有遗漏任何复制的列，因为这可能导致潜在的
		 * 不必要的数据丢失。
		 */
		fc_desc = RelationGetDescr(fc_entry->localrel);
		fc_oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
		fc_entry->attrmap = make_attrmap(fc_desc->natts);
		MemoryContextSwitchTo(fc_oldctx);

		/* 检查并报告缺失的属性，如果有的话 */
		fc_missingatts = bms_add_range(NULL, 0, fc_remoterel->natts - 1);
		for (fc_i = 0; fc_i < fc_desc->natts; fc_i++)
		{
			int			fc_attnum;
			Form_pg_attribute fc_attr = TupleDescAttr(fc_desc, fc_i);

			if (fc_attr->attisdropped || fc_attr->attgenerated)
			{
				fc_entry->attrmap->attnums[fc_i] = -1;
				continue;
			}

			fc_attnum = fc_logicalrep_rel_att_by_name(fc_remoterel,
												NameStr(fc_attr->attname));

			fc_entry->attrmap->attnums[fc_i] = fc_attnum;
			if (fc_attnum >= 0)
				fc_missingatts = bms_del_member(fc_missingatts, fc_attnum);
		}

		fc_logicalrep_report_missing_attrs(fc_remoterel, fc_missingatts);

		/* 保持整洁 */
		bms_free(fc_missingatts);

		/*
		 * 设置如果表的复制标识足够用于应用
		 * 更新/删除。
		 */
		fc_logicalrep_rel_mark_updatable(fc_entry);

		fc_entry->localrelvalid = true;
	}

	if (fc_entry->state != SUBREL_STATE_READY)
		fc_entry->state = GetSubscriptionRelState(MySubscription->oid,
											   fc_entry->localreloid,
											   &fc_entry->statelsn);

	return fc_entry;
}

/*
 * 关闭之前打开的逻辑关系。
 */
void logicalrep_rel_close(LogicalRepRelMapEntry *fc_rel, LOCKMODE fc_lockmode)
{
	table_close(fc_rel->localrel, fc_lockmode);
	fc_rel->localrel = NULL;
}

/*
 * 分区缓存：查找分区 LogicalRepRelMapEntry 的
 *
 * 与关系映射缓存不同，这个是以分区 OID 为键，而不是远程
 * 关系 OID，因为我们只需在分区未直接映射到任何远程关系的情况下使用此缓存，
 * 例如当复制发生在其祖先作为目标时。
 */

/*
 * Relcache 无效化回调
 */
static void fc_logicalrep_partmap_invalidate_cb(Datum fc_arg, Oid fc_reloid)
{
	LogicalRepPartMapEntry *fc_entry;

	/* 只是为了确保。 */
	if (LogicalRepPartMap == NULL)
		return;

	if (fc_reloid != InvalidOid)
	{
		HASH_SEQ_STATUS fc_status;

		hash_seq_init(&fc_status, LogicalRepPartMap);

		/* TODO，使用逆查找哈希表？ */
		while ((fc_entry = (LogicalRepPartMapEntry *) hash_seq_search(&fc_status)) != NULL)
		{
			if (fc_entry->relmapentry.localreloid == fc_reloid)
			{
				fc_entry->relmapentry.localrelvalid = false;
				hash_seq_term(&fc_status);
				break;
			}
		}
	}
	else
	{
		/* 使所有缓存条目失效 */
		HASH_SEQ_STATUS fc_status;

		hash_seq_init(&fc_status, LogicalRepPartMap);

		while ((fc_entry = (LogicalRepPartMapEntry *) hash_seq_search(&fc_status)) != NULL)
			fc_entry->relmapentry.localrelvalid = false;
	}
}

/*
 * 重置在分区映射中指向 remoterel 的条目。
 *
 * 当发布者发送新的关系映射以更新我们对来自该发布者的
 * 传入数据的预期视图时调用。
 *
 * 注意我们在此处不会更新条目的 remoterel 信息，
 * 我们将在 logicalrep_partition_open 中更新信息以避免
 * 不必要的工作。
 */
void logicalrep_partmap_reset_relmap(LogicalRepRelation *fc_remoterel)
{
	HASH_SEQ_STATUS fc_status;
	LogicalRepPartMapEntry *fc_part_entry;
	LogicalRepRelMapEntry *fc_entry;

	if (LogicalRepPartMap == NULL)
		return;

	hash_seq_init(&fc_status, LogicalRepPartMap);
	while ((fc_part_entry = (LogicalRepPartMapEntry *) hash_seq_search(&fc_status)) != NULL)
	{
		fc_entry = &fc_part_entry->relmapentry;

		if (fc_entry->remoterel.remoteid != fc_remoterel->remoteid)
			continue;

		fc_logicalrep_relmap_free_entry(fc_entry);

		memset(fc_entry, 0, sizeof(LogicalRepRelMapEntry));
	}
}

/*
 * 初始化分区映射缓存。
 */
static void fc_logicalrep_partmap_init(void)
{
	HASHCTL		fc_ctl;

	if (!LogicalRepPartMapContext)
		LogicalRepPartMapContext =
			AllocSetContextCreate(CacheMemoryContext,
								  "LogicalRepPartMapContext",
								  ALLOCSET_DEFAULT_SIZES);

	/* 初始化关系哈希表。 */
	fc_ctl.keysize = sizeof(Oid);	/* 分区 OID */
	fc_ctl.entrysize = sizeof(LogicalRepPartMapEntry);
	fc_ctl.hcxt = LogicalRepPartMapContext;

	LogicalRepPartMap = hash_create("logicalrep partition map cache", 64, &fc_ctl,
									HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);

	/* 监视失效事件。 */
	CacheRegisterRelcacheCallback(fc_logicalrep_partmap_invalidate_cb,
								  (Datum) 0);
}

/*
 * logicalrep_partition_open
 *
 * 返回的条目重用的值大部分来自根表的条目，保存
 * 属性映射，因为对于分区可能会有所不同。然而，
 * 我们必须物理复制所有数据，以防根表的条目
 * 被释放/重建。
 *
 * 注意这里没有 logicalrep_partition_close，因为调用者关闭
 * 组件关系。
 */
LogicalRepRelMapEntry * logicalrep_partition_open(LogicalRepRelMapEntry *fc_root,
						  Relation fc_partrel, AttrMap *fc_map)
{
	LogicalRepRelMapEntry *fc_entry;
	LogicalRepPartMapEntry *fc_part_entry;
	LogicalRepRelation *fc_remoterel = &fc_root->remoterel;
	Oid			fc_partOid = RelationGetRelid(fc_partrel);
	AttrMap    *fc_attrmap = fc_root->attrmap;
	bool		fc_found;
	MemoryContext fc_oldctx;

	if (LogicalRepPartMap == NULL)
		fc_logicalrep_partmap_init();

	/* 搜索现有条目。 */
	fc_part_entry = (LogicalRepPartMapEntry *) hash_search(LogicalRepPartMap,
														(void *) &fc_partOid,
														HASH_ENTER, &fc_found);

	fc_entry = &fc_part_entry->relmapentry;

	/*
	 * 我们必须始终用最新的分区
	 * Relation 指针覆盖 entry->localrel，因为旧值指向的关系
	 * 可能在调用者最后一次使用此条目后关闭分区
	 * 关系后被清除。注意 localrelvalid 仅通过
	 * relcache 无效化回调更新，因此无论 localrel 指向的
	 * 关系是否已被清除，它可能仍然为真。
	 */
	if (fc_found && fc_entry->localrelvalid)
	{
		fc_entry->localrel = fc_partrel;
		return fc_entry;
	}

	/* 切换到生命周期更长的上下文。 */
	fc_oldctx = MemoryContextSwitchTo(LogicalRepPartMapContext);

	if (!fc_found)
	{
		memset(fc_part_entry, 0, sizeof(LogicalRepPartMapEntry));
		fc_part_entry->partoid = fc_partOid;
	}

	/* 释放不再有用的 attrmap（属性映射），如果有的话。 */
	if (fc_entry->attrmap)
	{
		free_attrmap(fc_entry->attrmap);
		fc_entry->attrmap = NULL;
	}

	if (!fc_entry->remoterel.remoteid)
	{
		int			fc_i;

		/* 远程关系直接从根条目复制。 */
		fc_entry = &fc_part_entry->relmapentry;
		fc_entry->remoterel.remoteid = fc_remoterel->remoteid;
		fc_entry->remoterel.nspname = pstrdup(fc_remoterel->nspname);
		fc_entry->remoterel.relname = pstrdup(fc_remoterel->relname);
		fc_entry->remoterel.natts = fc_remoterel->natts;
		fc_entry->remoterel.attnames = palloc(fc_remoterel->natts * sizeof(char *));
		fc_entry->remoterel.atttyps = palloc(fc_remoterel->natts * sizeof(Oid));
		for (fc_i = 0; fc_i < fc_remoterel->natts; fc_i++)
		{
			fc_entry->remoterel.attnames[fc_i] = pstrdup(fc_remoterel->attnames[fc_i]);
			fc_entry->remoterel.atttyps[fc_i] = fc_remoterel->atttyps[fc_i];
		}
		fc_entry->remoterel.replident = fc_remoterel->replident;
		fc_entry->remoterel.attkeys = bms_copy(fc_remoterel->attkeys);
	}

	fc_entry->localrel = fc_partrel;
	fc_entry->localreloid = fc_partOid;

	/*
	 * 如果分区的属性与根关系的属性不匹配，我们需要
	 * 创建一个新的属性映射，将分区属性编号映射到
	 * remoterel，而不是原来的将根关系的
	 * 属性编号映射到remoterel。
	 *
	 * 注意，从元组路由数据结构中得到的'map'包含的是1基数属性编号
	 * （父关系的）。然而，在'entry'中，逻辑复制数据结构包含的是
	 * 0基数属性编号（远程关系的）。
	 */
	if (fc_map)
	{
		AttrNumber	fc_attno;

		fc_entry->attrmap = make_attrmap(fc_map->maplen);
		for (fc_attno = 0; fc_attno < fc_entry->attrmap->maplen; fc_attno++)
		{
			AttrNumber	fc_root_attno = fc_map->attnums[fc_attno];

			/* 0 表示这是一个已删除的属性。请参见AttrMap顶部的注释。 */
			if (fc_root_attno == 0)
				fc_entry->attrmap->attnums[fc_attno] = -1;
			else
				fc_entry->attrmap->attnums[fc_attno] = fc_attrmap->attnums[fc_root_attno - 1];
		}
	}
	else
	{
		/* 缺少copy_attmap，采用困难的方法。 */
		fc_entry->attrmap = make_attrmap(fc_attrmap->maplen);
		memcpy(fc_entry->attrmap->attnums, fc_attrmap->attnums,
			   fc_attrmap->maplen * sizeof(AttrNumber));
	}

	/* 如果表的副本身份足以应用更新/删除，则设置此项。 */
	fc_logicalrep_rel_mark_updatable(fc_entry);

	fc_entry->localrelvalid = true;

	/* state和statelsn保持为0。 */
	MemoryContextSwitchTo(fc_oldctx);

	return fc_entry;
}

#ifdef FDDM
//重写logicalrep_rel_open，用于fdd_mmr捕获table_missing错误

/*
 * 打开与远程关系关联的本地关系。
 *
 * 如果由于本地DDL而使Relcache映射失效，则重建该映射。
 */
LogicalRepRelMapEntry * fdd_logicalrep_rel_open(LogicalRepRelId fc_remoteid, LOCKMODE fc_lockmode,LogicalRepTupleData *fc_tup, bool *fc_skip, column_missing_conflict_hook_type fc_column_missng_hook)
{
	LogicalRepRelMapEntry *fc_entry;
	bool		fc_found;
	LogicalRepRelation *fc_remoterel;

	if (LogicalRepRelMap == NULL)
		fc_logicalrep_relmap_init();

	/* 搜索现有条目。 */
	fc_entry = hash_search(LogicalRepRelMap, (void *) &fc_remoteid,
						HASH_FIND, &fc_found);

	if (!fc_found)
	{
		//这里是流程有问题，没法算table_missing
		elog(ERROR, "no relation map entry for remote relation ID %u",
			 fc_remoteid);
	}


	fc_remoterel = &fc_entry->remoterel;

	/* 确保我们不会泄漏relcache引用计数。 */
	if (fc_entry->localrel)
		elog(ERROR, "remote relation ID %u is already open", fc_remoteid);

	/*
	 * 在打开和锁定关系时，处理待处理的失效消息，这可能会使关系失效。
	 * 因此，如果条目当前被认为是有效的，尝试通过OID打开本地关系并
	 * 查看是否发生失效。
	 */
	if (fc_entry->localrelvalid)
	{
		fc_entry->localrel = try_table_open(fc_entry->localreloid, fc_lockmode);
		if (!fc_entry->localrel)
		{
			/* 表被重命名或删除。 */
			fc_entry->localrelvalid = false;
		}
		else if (!fc_entry->localrelvalid)
		{
			/* 请注意，我们在这里释放不再有用的锁。 */
			table_close(fc_entry->localrel, fc_lockmode);
			fc_entry->localrel = NULL;
		}
	}
	//走到这里，已经确定localrelvalid为true，并且上一步打开了本地表，需要再次检查列缺失情况
	//如果localrelvalid为false，会在后面再去打开本地表的时候检查
	if (fc_entry->localrelvalid)
	{
		//如果上一次列缺失，但是选择了继续处理，这里的表已经打开成功，要判断一下
		//一是为了记录冲突，二是如果冲突处理方式改变了，要进行相应的处理
		bool fc_skip_for_column_miss = false;
		TupleDesc fc_desc = RelationGetDescr(fc_entry->localrel);
		fc_skip_for_column_miss = fc_column_missng_hook(fc_remoterel, fc_tup, fc_desc, fc_entry);
		if (fc_skip_for_column_miss)
		{
			*fc_skip = true;
			table_close(fc_entry->localrel, NoLock);
			fc_entry->localrel = NULL;
			//将表映射改为无效
			fc_entry->localrelvalid = false;
			return fc_entry;
		}
	}

	/*
	 * 如果自上次对条目加锁以来条目已被标记为无效，
	 * 按名称重新打开本地关系并重建所有派生数据。
	 */
	if (!fc_entry->localrelvalid)
	{
		Oid			fc_relid;
		Bitmapset  *fc_idkey;
		TupleDesc	fc_desc;
		MemoryContext fc_oldctx;
		int			fc_i;
		bool fc_skip_for_column_miss = false;
		//Bitmapset  *missingatts;

		/* 尝试通过名称查找并锁定关系。 */
		fc_relid = RangeVarGetRelid(makeRangeVar(fc_remoterel->nspname,
											  fc_remoterel->relname, -1),
								 fc_lockmode, true);
		if (!OidIsValid(fc_relid))
		{
			//table_miss conflict
			return fc_entry;
		}
		fc_entry->localrel = table_open(fc_relid, NoLock);
		fc_entry->localreloid = fc_relid;

		/* 检查支持的 relkind。 */
		CheckSubscriptionRelkind(fc_entry->localrel->rd_rel->relkind,
								 fc_remoterel->nspname, fc_remoterel->relname);

		/*
		 * 构建本地属性编号到远程属性编号的映射
		 * 并验证我们没有遗漏任何复制的列，因为这可能导致潜在的
		 * 不必要的数据丢失。
		 */
		fc_desc = RelationGetDescr(fc_entry->localrel);
		fc_oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
		fc_entry->attrmap = make_attrmap(fc_desc->natts);
		MemoryContextSwitchTo(fc_oldctx);

	
		//column missing conflict handle
		fc_skip_for_column_miss = fc_column_missng_hook(fc_remoterel, fc_tup, fc_desc, fc_entry);
		if (fc_skip_for_column_miss)
		{
			//这里表映射还是无效的
			*fc_skip = true;
			table_close(fc_entry->localrel, NoLock);
			fc_entry->localrel = NULL;
			return fc_entry;
		}
	
		/*
		 * 检查副本身份是否匹配。我们允许在订阅者上有更严格的副本身份
		 * （较少的列），因为这不会阻止我们找到唯一的元组。即如果发布者有身份
		 * （id,timestamp），而订阅者只有（id），这将不是问题，
		 * 但在相反的情况下会是。
		 *
		 * 不要在这里抛出任何错误，只需将关系条目标记为不可更新，因为副本身份
		 * 仅适用于更新和删除，但即使没有它，插入也可以被复制。
		 */
		fc_entry->updatable = true;
		fc_idkey = RelationGetIndexAttrBitmap(fc_entry->localrel,
										   INDEX_ATTR_BITMAP_IDENTITY_KEY);
		/* 如果没有副本身份，则回退到PK */
		if (fc_idkey == NULL)
		{
			fc_idkey = RelationGetIndexAttrBitmap(fc_entry->localrel,
											   INDEX_ATTR_BITMAP_PRIMARY_KEY);

			/*
			 * 如果没有副本身份索引且没有主键，则发布的表
			 * 必须具有副本身份FULL。
			 */
			if (fc_idkey == NULL && fc_remoterel->replident != REPLICA_IDENTITY_FULL)
				fc_entry->updatable = false;
		}

		fc_i = -1;
		while ((fc_i = bms_next_member(fc_idkey, fc_i)) >= 0)
		{
			int			fc_attnum = fc_i + FirstLowInvalidHeapAttributeNumber;

			if (!AttrNumberIsForUserDefinedAttr(fc_attnum))
			{
				ereport(ERROR,
						(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
						 errmsg("logical replication target relation \"%s.%s\" uses "
								"system columns in REPLICA IDENTITY index",
								fc_remoterel->nspname, fc_remoterel->relname)));
			}

			fc_attnum = AttrNumberGetAttrOffset(fc_attnum);

			if (fc_entry->attrmap->attnums[fc_attnum] < 0 ||
				!bms_is_member(fc_entry->attrmap->attnums[fc_attnum], fc_remoterel->attkeys))
			{
				fc_entry->updatable = false;
				break;
			}
		}

		fc_entry->localrelvalid = true;
	}

	if (fc_entry->state != SUBREL_STATE_READY)
		fc_entry->state = GetSubscriptionRelState(MySubscription->oid,
											   fc_entry->localreloid,
											   &fc_entry->statelsn);

	return fc_entry;
}
#endif

